#include "stdafx.h"
#include "net/SocketReactor.hpp"
#include "net/SocketNotifier.hpp"
#include "thread/errorhandler.hpp"
using JHCPP::thread::CErrorHandler;
#pragma warning(disable:4355)

NET_NAMESPACE_BEGIN

CSocketReactor::CSocketReactor()
	: m_stop(false),
	m_timeout(DEFAULT_TIMEOUT),
	m_pReadableNotification(new CReadableNotification(this)),
	m_pWritableNotification(new CWritableNotification(this)),
	m_pErrorNotification(new CErrorNotification(this)),
	m_pTimeoutNotification(new CTimeoutNotification(this)),
	m_pIdleNotification(new CIdleNotification(this)),
	m_pShutdownNotification(new CShutdownNotification(this)),
	m_pThread(0)
{
}

CSocketReactor::CSocketReactor(const CTimeSpan& timeout)
	: m_stop(false),
	m_timeout(timeout),
	m_pReadableNotification(new CReadableNotification(this)),
	m_pWritableNotification(new CWritableNotification(this)),
	m_pErrorNotification(new CErrorNotification(this)),
	m_pTimeoutNotification(new CTimeoutNotification(this)),
	m_pIdleNotification(new CIdleNotification(this)),
	m_pShutdownNotification(new CShutdownNotification(this)),
	m_pThread(0)
{
}

CSocketReactor::~CSocketReactor()
{
}

void CSocketReactor::run()
{
	m_pThread = CThread::current();

	CSocket::SocketList readable;
	CSocket::SocketList writable;
	CSocket::SocketList except;

	while (!m_stop)
	{
		try
		{
			readable.clear();
			writable.clear();
			except.clear();
			int nSockets = 0;
			{
				CMutex::ScopedLock lock(m_mutex);
				for (EventHandlerMap::iterator it = m_handlers.begin(); it != m_handlers.end(); ++it)
				{
					if (it->second->accepts(m_pReadableNotification))
					{
						readable.push_back(it->first);
						nSockets++;
					}
					if (it->second->accepts(m_pWritableNotification))
					{
						writable.push_back(it->first);
						nSockets++;
					}
					if (it->second->accepts(m_pErrorNotification))
					{
						except.push_back(it->first);
						nSockets++;
					}
				}
			}
			if (nSockets == 0)
			{
				onIdle();
				CThread::trySleep(m_timeout.milliseconds());
			}
			else if (CSocket::select(readable, writable, except, m_timeout))
			{
				onBusy();

				for (CSocket::SocketList::iterator it = readable.begin(); it != readable.end(); ++it)
					dispatch(*it, m_pReadableNotification);
				for (CSocket::SocketList::iterator it = writable.begin(); it != writable.end(); ++it)
					dispatch(*it, m_pWritableNotification);
				for (CSocket::SocketList::iterator it = except.begin(); it != except.end(); ++it)
					dispatch(*it, m_pErrorNotification);
			}
			else onTimeout();
		}
		catch (CException& exc)
		{
			CErrorHandler::handle(exc);
		}
		catch (std::exception& exc)
		{
			CErrorHandler::handle(exc);
		}
		catch (...)
		{
			CErrorHandler::handle();
		}
	}
	onShutdown();
}

void CSocketReactor::stop()
{
	m_stop = true;
}

void CSocketReactor::wakeUp()
{
	if (m_pThread) 
		m_pThread->wakeUp();
}

void CSocketReactor::setTimeout(const CTimeSpan& timeout)
{
	m_timeout = timeout;
}

const CTimeSpan& CSocketReactor::getTimeout() const
{
	return m_timeout;
}

void CSocketReactor::addEventHandler(const CSocket& socket, const CAbstractObserver& observer)
{
	NotifierPtr pNotifier;
	{
		CMutex::ScopedLock lock(m_mutex);

		EventHandlerMap::iterator it = m_handlers.find(socket);
		if (it == m_handlers.end())
		{
			pNotifier = new CSocketNotifier(socket);
			m_handlers[socket] = pNotifier;
		}
		else pNotifier = it->second;
	}
	if (!pNotifier->hasObserver(observer))
		pNotifier->addObserver(this, observer);
}

bool CSocketReactor::hasEventHandler(const CSocket& socket, const CAbstractObserver& observer)
{
	NotifierPtr pNotifier;
	{
		CMutex::ScopedLock lock(m_mutex);

		EventHandlerMap::iterator it = m_handlers.find(socket);
		if (it != m_handlers.end())
		{
			if (it->second->hasObserver(observer))
				return true;
		}
	}

	return false;
}

void CSocketReactor::removeEventHandler(const CSocket& socket, const CAbstractObserver& observer)
{
	NotifierPtr pNotifier;
	{
		CMutex::ScopedLock lock(m_mutex);

		EventHandlerMap::iterator it = m_handlers.find(socket);
		if (it != m_handlers.end())
		{
			pNotifier = it->second;
			if (pNotifier->hasObserver(observer) && pNotifier->countObservers() == 1)
			{
				m_handlers.erase(it);
			}
		}
	}
	if (pNotifier && pNotifier->hasObserver(observer))
	{
		pNotifier->removeObserver(this, observer);
	}

}

void CSocketReactor::onTimeout()
{
	dispatch(m_pTimeoutNotification);
}

void CSocketReactor::onIdle()
{
	dispatch(m_pIdleNotification);
}

void CSocketReactor::onShutdown()
{
	dispatch(m_pShutdownNotification);
}

void CSocketReactor::onBusy()
{
}

void CSocketReactor::dispatch(const CSocket& socket, CSocketNotification* pNotification)
{
	NotifierPtr pNotifier;
	{
		CMutex::ScopedLock lock(m_mutex);
		EventHandlerMap::iterator it = m_handlers.find(socket);
		if (it != m_handlers.end())
			pNotifier = it->second;
		else
			return;
	}
	dispatch(pNotifier, pNotification);
}

void CSocketReactor::dispatch(CSocketNotification* pNotification)
{
	std::vector<NotifierPtr> delegates;
	delegates.reserve(m_handlers.size());
	{
		CMutex::ScopedLock lock(m_mutex);
		for (EventHandlerMap::iterator it = m_handlers.begin(); it != m_handlers.end(); ++it)
			delegates.push_back(it->second);
	}
	for (std::vector<NotifierPtr>::iterator it = delegates.begin(); it != delegates.end(); ++it)
	{
		dispatch(*it, pNotification);
	}
}

void CSocketReactor::dispatch(NotifierPtr& pNotifier, CSocketNotification* pNotification)
{
	try
	{
		pNotifier->dispatch(pNotification);
	}
	catch (CException& exc)
	{
		CErrorHandler::handle(exc);
	}
	catch (std::exception& exc)
	{
		CErrorHandler::handle(exc);
	}
	catch (...)
	{
		CErrorHandler::handle();
	}
}

NET_NAMESPACE_END
